package com.heima;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //2.封装kafka相关信息
        Properties config = new Properties();
        //3.指定kafka地址,setProperty只接受string类型的数据
        config.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.225.128:9092,192.168.225.128:9192");
        //指定消息的序列化
        config.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        config.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //设置ack的类型
        config.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        //设置消息重发次数
        config.setProperty(ProducerConfig.RETRIES_CONFIG, "10");
        //消息压缩
        config.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        //4.指定消息key和value的序列化器（转化为字节数组）
        //1.创建生产者对象，，kafka 中消息格式key value格式
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);
        //5.1 指定发送的主题，key，消息体
        ProducerRecord<String, String> producerRecord1 = new ProducerRecord<String, String>("TopicSource", 0,"369", "动不动拉扯啊");
        ProducerRecord<String, String> producerRecord2 = new ProducerRecord<String, String>("TopicSource", 0,"369", "动不动拉扯啊");
        ProducerRecord<String, String> producerRecord3 = new ProducerRecord<String, String>("TopicSource", 1,"369", "动不动拉扯啊");
        ProducerRecord<String, String> producerRecord4 = new ProducerRecord<String, String>("TopicSource", 1,"369", "动不动拉扯啊");
        //5.生产者发送消息
        RecordMetadata recordMetadata1 = producer.send(producerRecord1).get();
        RecordMetadata recordMetadata2 = producer.send(producerRecord2).get();
        RecordMetadata recordMetadata3 = producer.send(producerRecord3).get();
        RecordMetadata recordMetadata4 = producer.send(producerRecord4).get();

        System.out.println("1发送到的分区为"+recordMetadata1.partition());
        System.out.println("2发送到的分区为"+recordMetadata2.partition());
        System.out.println("3发送到的分区为"+recordMetadata3.partition());
        System.out.println("4发送到的分区为"+recordMetadata4.partition());
         producer.send(producerRecord4, (recordMetadata, e) -> {
             int partition = recordMetadata.partition();
             long offset = recordMetadata.offset();
             System.out.println("异步消息发送回调，分区为："+partition);
             System.out.println("异步消息发送回调，偏移量为："+offset);
         });
        //6.关闭消息通道
        producer.close();
        System.out.println("消息发送成功：" + producerRecord1);
        System.out.println("消息发送成功：" + producerRecord2);
        System.out.println("消息发送成功：" + producerRecord3);
        System.out.println("消息发送成功：" + producerRecord4);
    }
}
